package fi.aalto.cse.T110_5150.P2PClient;

import java.io.*;
import java.net.*;
import java.util.*;

public class LocalPeer implements Runnable {

	private static LocalPeer instance;

	/**
	 * Private variables
	 */
	private ServerSocket server;
	private Hashtable<PeerAddress, RemotePeer> remotePeers;
	private PeerAddress address;
	private List<PeerAddress> randomPeerAddresses;
	private SharedResources sharedResources;
	private QueryHistory queryHistory;
	private boolean isInitialized;
	private boolean autoJoin;
	private boolean listen;
	private Thread listeningThread;

	public LocalPeer() throws IOException {
		address = new PeerAddress(PeerAddress.getLocalHost(), 0);
		instance = this;
		sharedResources = new SharedResources();
		queryHistory = new QueryHistory();
		remotePeers = new Hashtable<PeerAddress, RemotePeer>();
		randomPeerAddresses = new ArrayList<PeerAddress>();
		setAutoJoin(true);
		setListen(true);
	}

	public static LocalPeer getInstance() {
		return instance;
	}

	public void initialize(int port) throws P2PException {
		if (isInitialized) {
			Logger.error("The local peer is already initialized.");
			return;
		}

		try {
			address.setPort(port);

			server = new ServerSocket(port);

			Logger.info("STARTED");

			isInitialized = true;

			setListen(listen);

		} catch (IOException e) {
			String message = String.format("Could not start on port %d", address.getPort());
			throw new P2PException(message, e);
		}
	}

	/**
	 * Public methods
	 */

	public boolean getInitialized() {
		return isInitialized;
	}

	@Override
	public String toString() {
		return address.toString();
	}

	public PeerAddress getAddress() {
		return address;
	}

	public Hashtable<PeerAddress, RemotePeer> getRemotePeers() {
		return remotePeers;
	}

	public SharedResources getSharedResources() {
		return sharedResources;
	}

	public void setAutoJoin(boolean autoJoin) {
		this.autoJoin = autoJoin;
	}

	public boolean getAutoJoin() {
		return autoJoin;
	}

	public void setListen(boolean listen) {
		this.listen = listen;

		if (isInitialized && listen && listeningThread == null) {
			listeningThread = new Thread(this);
			listeningThread.start();
		}
	}

	public boolean getListen() {
		return listen;
	}

	@Override
	public void run() {
		synchronized (listeningThread) {
			try {
				server.setSoTimeout(100);
			} catch (SocketException e) {
				Logger.error("Cannot set server socket timeout", e);
				return;
			}
			while (getListen()) {
				try {
					Socket socket = server.accept();
					if (!socket.isClosed()) {
						PeerAddress remotePeerAddress = new PeerAddress(socket.getInetAddress(), socket.getPort());
						RemotePeer remotePeer = new RemotePeer(socket, remotePeerAddress, false);

						Thread thread = new Thread(remotePeer);
						thread.start();
					}
				} catch (SocketTimeoutException e) {
				} catch (IOException e) {
					String message = String.format("Could not start P2P peer on port %d.%n", address.getPort());
					Logger.error(new P2PException(message, e));
				}
			}
			Logger.warning("Stopped listening");
			listeningThread = null;
		}
	}

	public void stop() {
		setListen(false);
	}

	@Override
	protected void finalize() {

		try {

			super.finalize();

			if (server != null) {

				for (Iterator<RemotePeer> it = remotePeers.values().iterator(); it.hasNext();) {
					RemotePeer remotePeer = it.next();
					remotePeer.stop(true);
				}

				remotePeers.clear();

				server.close();
			}

		} catch (Throwable e) {
			Logger.error("Unexpected error", e);
		} finally {
			server = null;
		}
	}

	public void connect(PeerAddress remotePeerAddress) throws P2PException {
		if (!isInitialized) {
			throw new P2PException("The local peer is not initialized.");
		}

		if (remotePeers.containsKey(remotePeerAddress)) {
			throw new P2PAlreadyConnectedException(String.format("Already connected with '%s'", remotePeerAddress));
		}

		try {
			Socket socket = new Socket(remotePeerAddress.getAddress(), remotePeerAddress.getPort());
			RemotePeer remotePeer = new RemotePeer(socket, remotePeerAddress, true);

			Thread thread = new Thread(remotePeer);
			thread.start();

			addRemotePeer(remotePeer);
		} catch (IOException e) {
			String message = String.format("Could not connect to %s: %s", remotePeerAddress, e.getMessage());
			throw new P2PException(message);
		}
	}

	public void disconnect(PeerAddress remotePeerAddress) throws P2PException {
		if (!isInitialized) {
			throw new P2PException("The local peer is not initialized.");
		}

		RemotePeer remotePeer = remotePeers.get(remotePeerAddress);

		if (remotePeer != null) {
			remotePeer.stop(true);
		} else {
			throw new P2PException(String.format("Peer '%s' is not found", remotePeerAddress));
		}
	}

	public void sendMessageQuery(ByteArray queryKey) throws P2PException {
		if (!isInitialized) {
			Logger.error("The local peer is not initialized.");
			return;
		}

		if (queryKey.size() > Short.MAX_VALUE)
			throw new P2PException(String.format("Query length should be less than %d characters", Short.MAX_VALUE));

		int count = 0;

		MessageHeader header = MessageHeader.create(P2PHelper.MESSAGE_TYPE_QUERY, P2PHelper.QUERY_MAX_TTL,
				queryKey.size());

		for (Iterator<RemotePeer> it = remotePeers.values().iterator(); it.hasNext();) {
			RemotePeer remotePeer = it.next();
			if (remotePeer.getInitializedStatus()) {
				try {
					remotePeer.sendMessageQuery(header, queryKey);
				} catch (IOException e) {
					throw new P2PException(String.format("I/O error: %s", e.getMessage()), e);
				}
				count++;
				Logger.debugOutputAction(Logger.QUERY, "Sent Query Message", remotePeer.getAddress());
			}
		}

		queryHistory.addQuery(header.messageId, queryKey, null);
		Logger.info(String.format("Sent query messages to %d peers", count));
	}

	public void handleMessageQuery(MessageHeader header, ByteArray queryKey, RemotePeer remotePeer) throws IOException {
		PeerAddress remotePeerAddress = remotePeer.getAddress();

		/* check if same message with in a given amount of time */
		if (!queryHistory.tryAddQuery(header.messageId, queryKey, remotePeer)) {
			Logger.debugInputAction(Logger.QUERY,
					String.format("Received duplicate query (message ID=%d)", header.messageId), remotePeerAddress);
			return;
		}

		/* 1. Check for match in local shared files */
		Resource[] resources = sharedResources.get(queryKey);

		if (resources != null) {
			remotePeer.sendMessageQueryHit(header.messageId, resources);

			Logger.infoOutputAction(String.format("Sent QueryHit Message for key '%s'", queryKey), remotePeerAddress);

		} else {
			Logger.infoInputAction(String.format("Received and ignored Query Message (key '%s' not found)", queryKey),
					remotePeerAddress);
		}

		/* 2. Forward message to all the peers */
		if (header.ttl > 0) {
			PeerAddress originalPeerAddress = header.getOriginalPeerAddress();

			header.ttl--;

			int count = 0;

			for (Enumeration<RemotePeer> en = remotePeers.elements(); en.hasMoreElements();) {

				RemotePeer forwardRemotePeer = en.nextElement();
				PeerAddress forwardRemotePeerAddress = forwardRemotePeer.getAddress();

				if (!forwardRemotePeerAddress.equals(remotePeerAddress)
						&& !forwardRemotePeerAddress.equals(originalPeerAddress)) {
					forwardRemotePeer.sendMessageQuery(header, queryKey);
					Logger.debugOutputAction(Logger.QUERY, "Forwarded Query Message", forwardRemotePeerAddress);
					count++;
				}
			}

			Logger.info(String.format("Forwarded Query Message to %d other peers", count));
		}
	}

	public void handleMessageQueryHit(MessageHeader header, Resource[] resources) throws IOException {

		QueryInfo query = queryHistory.getQuery(header.messageId);

		if (query != null) {
			RemotePeer sender = query.getSender();

			if (sender == null) { // the query came originally from the local
									// peer

				String message = String.format("Received %d values for key '%s' (", resources.length, query.getKey());

				for (int i = 0; i < resources.length; ++i) {
					message += String.format(" %s", resources[i]);
				}

				message += " )";

				Logger.infoInputAction(message, header.getOriginalPeerAddress());

			} else { // the query came from another peer

				sender.sendMessageQueryHit(header.messageId, resources);
			}

		} else {
			Logger.warning(String.format("Received invalid Query Hit (message ID = %d)", header.messageId));
		}

	}

	public List<PeerAddress> getRandomRemotePeerAddresses(PeerAddress currentRemotePeerAddress) {

		// get the number of peers to be selected
		int numberOfPeers = Math.min(P2PHelper.PING_PONG_NETWORK_PROBING_MAX_PEER, remotePeers.size() - 1);

		// create a new result array
		List<PeerAddress> result = new ArrayList<PeerAddress>();

		synchronized (remotePeers) {

			boolean currentRemotePeerAddressWasInRandomList = false;

			currentRemotePeerAddressWasInRandomList = randomPeerAddresses.remove(currentRemotePeerAddress);

			if (randomPeerAddresses.size() <= numberOfPeers) {

				result.addAll(randomPeerAddresses);

				randomPeerAddresses = new ArrayList<PeerAddress>();

				// add all remote peers
				for (Enumeration<PeerAddress> enumration = remotePeers.keys(); enumration.hasMoreElements();) {
					randomPeerAddresses.add(enumration.nextElement());
				}

				// remove all used remote peers
				randomPeerAddresses.removeAll(result);

				currentRemotePeerAddressWasInRandomList = randomPeerAddresses.remove(currentRemotePeerAddress);
			}

			while (result.size() < numberOfPeers) {

				// select any item from the random list
				int i = P2PHelper.nextRandom(randomPeerAddresses.size());
				result.add(randomPeerAddresses.get(i));
				randomPeerAddresses.remove(i);
			}

			if (currentRemotePeerAddressWasInRandomList) {
				randomPeerAddresses.add(currentRemotePeerAddress);
			}

		} // synchronized

		return result;
	}

	public void addRemotePeer(RemotePeer remotePeer) throws P2PException {
		PeerAddress remotePeerAddress = remotePeer.getAddress();
		synchronized (remotePeers) {
			if (remotePeers.containsKey(remotePeerAddress)) {
				throw new P2PException("Duplicated remote peer address: " + remotePeerAddress);
			}
			remotePeers.put(remotePeerAddress, remotePeer);
			if (randomPeerAddresses != null) {
				randomPeerAddresses.add(remotePeerAddress);
			}
		}
	}

	public void removeRemotePeer(RemotePeer remotePeer) throws P2PException {
		PeerAddress remotePeerAddress = remotePeer.getAddress();
		synchronized (remotePeers) {
			if (remotePeers.remove(remotePeerAddress) != null) {
				if (randomPeerAddresses != null) {
					randomPeerAddresses.remove(remotePeerAddress);
				}
			} else {
				throw new P2PException("Remote peer is not found: " + remotePeerAddress);
			}
		}
	}

}
